package org.nbict.iot.trash.mogo;

import org.apache.commons.lang.Validate;
import org.apache.storm.mongodb.common.mapper.MongoMapper;
import org.apache.storm.task.OutputCollector;
import org.apache.storm.task.TopologyContext;
import org.apache.storm.topology.OutputFieldsDeclarer;
import org.apache.storm.tuple.Tuple;
import org.apache.storm.utils.BatchHelper;
import org.apache.storm.utils.TupleUtils;
import org.bson.Document;

import java.util.Iterator;
import java.util.LinkedList;
import java.util.Map;

/**
 * Created by songseven on 18/6/20.
 */
@Deprecated
public class MongoInsertBolt extends AbstractMongoBolt {

    private static final int DEFAULT_FLUSH_INTERVAL_SECS = 1;
    private MongoMapper mapper;
    private boolean ordered = true;
    private int batchSize;
    private BatchHelper batchHelper;
    private int flushIntervalSecs = 1;

    public MongoInsertBolt(String url, String username, String pasaword, String adminname, String database, String collectionName, MongoMapper mapper) {
        super(url, username, pasaword, adminname, database, collectionName);
        Validate.notNull(mapper, "MongoMapper can not be null");
        this.mapper = mapper;
    }

    public void execute(Tuple tuple) {
        try {
            if(this.batchHelper.shouldHandle(tuple)) {
                this.batchHelper.addBatch(tuple);
            }

            if(this.batchHelper.shouldFlush()) {
                this.flushTuples();
                this.batchHelper.ack();
            }
        } catch (Exception var3) {
            this.batchHelper.fail(var3);
        }

    }

    private void flushTuples() {
        LinkedList docs = new LinkedList();
        Iterator var2 = this.batchHelper.getBatchTuples().iterator();

        while(var2.hasNext()) {
            Tuple t = (Tuple)var2.next();
            Document doc = this.mapper.toDocument(t);
            docs.add(doc);
        }

        this.mongoClient.insert(docs, this.ordered);
    }

    public MongoInsertBolt withBatchSize(int batchSize) {
        this.batchSize = batchSize;
        return this;
    }

    public MongoInsertBolt withOrdered(boolean ordered) {
        this.ordered = ordered;
        return this;
    }

    public MongoInsertBolt withFlushIntervalSecs(int flushIntervalSecs) {
        this.flushIntervalSecs = flushIntervalSecs;
        return this;
    }

    public Map<String, Object> getComponentConfiguration() {
        return TupleUtils.putTickFrequencyIntoComponentConfig(super.getComponentConfiguration(), this.flushIntervalSecs);
    }

    public void prepare(Map stormConf, TopologyContext context, OutputCollector
            collector) {
        super.prepare(stormConf, context, collector);
        this.batchHelper = new BatchHelper(this.batchSize, collector);
    }

    public void declareOutputFields(OutputFieldsDeclarer declarer) {
    }
}
